package sgt.project;

import ltq.util.*;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Map;

public class MaxTenCartBolt extends BaseRichBolt {

    ProductCart []productCarts=new ProductCart[10];
    Connection conn=null;
    PreparedStatement ps = null;
    ResultSet rs = null;
    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
        conn= DatabaseUtil.getConnection();
        for (int i=0;i<10;i++){
            productCarts[i]=new ProductCart(i+"",0);
        }
    }

    @Override
    public void execute(Tuple input) {
        String productId = input.getString(0);
        Integer cartNum=input.getInteger(2);
        int productOfArrayIndex = findProductOfArray(productCarts,productId);
        if (productOfArrayIndex!=10){
            productCarts[productOfArrayIndex]=new ProductCart(productId,cartNum);
            refreshData(productId,cartNum);
            sout(productCarts);
            System.out.println();
        }else {
            int i = findMinCart(productCarts);
            if (cartNum>productCarts[i].cart_num){
                productCarts[i]=new ProductCart(productId,cartNum);
                refreshData(productId,cartNum);
                sout(productCarts);
                System.out.println();
            }
        }

//        int i = findMinCart(productCarts);
//        if (cart_num>productCarts[i].cart_num){
//            productCarts[i]=new ProductCart(product_id,cart_num);
//            refreshData(product_id,cart_num);
//            sout(productCarts);
//            System.out.println();
//        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
    static class ProductCart{
        public ProductCart(String product_id, int cart_num) {
            this.product_id = product_id;
            this.cart_num = cart_num;
        }

        String product_id;
        int cart_num;
    }
    public static int findMinCart(ProductCart [] productCarts){
        int min=productCarts[0].cart_num,k=0;
        for (int i=0;i<productCarts.length;i++){
            if (min>productCarts[i].cart_num){
                k=i;
            }
        }
        return k;
    }
    public static void sout(ProductCart [] productCarts){
        for (int i=0;i<10;i++){
            System.out.print(productCarts[i].product_id+"-"+productCarts[i].cart_num+".  ");
        }
    }
    public void refreshData(String product_id,int cart){

        try {
            ps= conn.prepareStatement("select * from max_ten_carts where product_id=?");
            ps.setObject(1,product_id);
            rs= ps.executeQuery();
            if (rs.next()){
                ps= conn.prepareStatement(
                        "update max_ten_carts " +
                                "set cart=? where product_id=?");
                ps.setObject(1,cart);
                ps.setObject(2,product_id);
                ps.executeUpdate();
            }
            else {
                ps= conn.prepareStatement("insert into max_ten_carts values (?,?)");
                ps.setObject(1,product_id);
                ps.setObject(2,cart);

                ps.executeUpdate();
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }

    }
    public void refreshData2(String product_id,int cart){

        try {
            ps= conn.prepareStatement("select * from max_ten_carts where product_id=?");
            ps.setObject(1,product_id);
            rs= ps.executeQuery();
            if (rs.next()){
                ps= conn.prepareStatement(
                        "update max_ten_carts " +
                                "set cart=? where product_id=?");
                ps.setObject(1,cart);
                ps.setObject(2,product_id);
                ps.executeUpdate();
            }
            else {
                ps= conn.prepareStatement("insert into max_ten_carts values (?,?)");
                ps.setObject(1,product_id);
                ps.setObject(2,cart);

                ps.executeUpdate();
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }

    }
    public static int findProductOfArray(ProductCart [] productCarts,String product_id){
        int i;
        for (i=0;i<10;i++){
            if (productCarts[i].product_id.equals(product_id)){
                break;
            }
        }
        return i;
    }
}
